Apache Hadoop Distributed Copy – DistCp Guide

DistCp (distributed copy) is a tool used for large inter/intra-cluster copying. It uses MapReduce to effect its distribution, error handling and recovery, and reporting. It expands a list of files and directories into input to map tasks, each of which will copy a partition of the files specified in the source list.

[The erstwhile implementation of DistCp] (http://hadoop.apache.org/docs/r1.2.1/distcp.html) has its share of quirks and drawbacks, both in its usage, as well as its extensibility and performance. The purpose of the DistCp refactor was to fix these shortcomings, enabling it to be used and extended programmatically. New paradigms have been introduced to improve runtime and setup performance, while simultaneously retaining the legacy behaviour as default.

This document aims to describe the design of the new DistCp, its spanking new features, their optimal use, and any deviance from the legacy implementation.

The most common invocation of DistCp is an inter-cluster copy:

This will expand the namespace under /foo/bar on nn1 into a temporary file, partition its contents among a set of map tasks, and start a copy on each NodeManager from nn1 to nn2.

One can also specify multiple source directories on the command line:

Or, equivalently, from a file using the -f option:

Where srclist contains

When copying from multiple sources, DistCp will abort the copy with an error message if two sources collide, but collisions at the destination are resolved per the options specified. By default, files already existing at the destination are skipped (i.e. not replaced by the source file). A count of skipped files is reported at the end of each job, but it may be inaccurate if a copier failed for some subset of its files, but succeeded on a later attempt.

It is important that each NodeManager can reach and communicate with both the source and destination file systems. For HDFS, both the source and destination must be running the same version of the protocol or use a backwards-compatible protocol; see [Copying Between Versions] (#Copying_Between_Versions_of_HDFS).

After a copy, it is recommended that one generates and cross-checks a listing of the source and destination to verify that the copy was truly successful. Since DistCp employs both Map/Reduce and the FileSystem API, issues in or between any of the three could adversely and silently affect the copy. Some have had success running with -update enabled to perform a second pass, but users should be acquainted with its semantics before attempting this.

It’s also worth noting that if another client is still writing to a source file, the copy will likely fail. Attempting to overwrite a file being written at the destination should also fail on HDFS. If a source file is (re)moved before it is copied, the copy will fail with a FileNotFoundException.

Please refer to the detailed Command Line Reference for information on all the options available in DistCp.

-update is used to copy files from source that don’t exist at the target or differ from the target version. -overwrite overwrites target-files that exist at the target.

The Update and Overwrite options warrant special attention since their handling of source-paths varies from the defaults in a very subtle manner. Consider a copy from /source/first/ and /source/second/ to /target/, where the source paths have the following contents:

When DistCp is invoked without -update or -overwrite, the DistCp defaults would create directories first/ and second/, under /target. Thus:

would yield the following contents in /target:

When either -update or -overwrite is specified, the contents of the source-directories are copied to target, and not the source directories themselves. Thus:

would yield the following contents in /target:

By extension, if both source folders contained a file with the same name (say, 0), then both sources would map an entry to /target/0 at the destination. Rather than to permit this conflict, DistCp will abort.

Now, consider the following copy operation:

With sources/sizes:

And destination/sizes:

Will effect:

1 is skipped because the file-length and contents match. 2 is copied because it doesn’t exist at the target. 10 and 20 are overwritten since the contents don’t match the source.

If -update is used, 1 is skipped because the file-length and contents match. 2 is copied because it doesn’t exist at the target. 10 and 20 are overwritten since the contents don’t match the source. However, if -append is additionally used, then only 10 is overwritten (source length less than destination) and 20 is appended with the change in file (if the files match up to the destination’s original length).

If -overwrite is used, 1 is overwritten as well.

-diff option syncs files from a source cluster to a target cluster with a snapshot diff. It copies, renames and removes files in the snapshot diff list.

-update option must be included when -diff option is in use.

Most cloud providers don’t work well with sync at the moment.

Usage:

Example:

The command above applies changes from snapshot snap1 to snap2 (i.e. snapshot diff from snap1 to snap2) in /src/ to /dst/. Obviously, it requires /src/ to have both snapshots snap1 and snap2. But the destination /dst/ must also have a snapshot with the same name as , in this case snap1. The destination /dst/ should not have new file operations (create, rename, delete) since snap1. Note that when this command finishes, a new snapshot snap2 will NOT be created at /dst/.

-update is required to use -diff option.

For instance, in /src/, if 1.txt is added and 2.txt is deleted after the creation of snap1 and before creation of snap2, the command above will copy 1.txt from /src/ to /dst/ and delete 2.txt from /dst/.

Sync behavior will be elaborated using experiments below.

Some preparations before we start.

Then we run distcp sync:

The command above should succeed. 1.txt will be copied from /src/ to /dst/. Again, -update option is required.

If we run the same command again, we will get DistCp sync failed exception because the destination has added a new file 1.txt since snap1. That being said, if we remove 1.txt manually from /dst/ and run the sync, the command will succeed.

First do a clean up from Experiment 1.

Run sync command, note the has been changed from snap2 in Experiment 1 to snap3.

Both 1.txt and 2.txt will be copied to /dst/.

Continuing from the end of Experiment 2:

Run sync command now:

2.txt is copied and 1.txt is deleted under /dst/.

Note that, though both /src/ and /dst/ have snapshot with the same name snap2, the snapshots don’t need to have the same content. That means, if you have a 1.txt in /dst/’s snap2 but they have different contents, 1.txt will still be removed from /dst/. The sync command doesn’t check the contents of the files that is going to be deleted. It simply follows the snapshot diff list between and .

Also, if we delete 1.txt from /dst/ before creating snap2 on /dst/ in the steps above, so that /dst/’s snap2 doesn’t have 1.txt before running sync command, the command will still succeed. It won’t throw exception while trying to delete 1.txt from /dst/ which doesn’t exist.

This section only applies to HDFS.

If the target and all of the source pathnames are in the /.reserved/raw hierarchy, then ‘raw’ namespace extended attributes will be preserved. ‘raw’ xattrs are used by the system for internal functions such as encryption meta data. They are only visible to users when accessed through the /.reserved/raw hierarchy.

raw xattrs are preserved based solely on whether /.reserved/raw prefixes are supplied. The -p (preserve, see below) flag does not impact preservation of raw xattrs.

To prevent raw xattrs from being preserved, simply do not use the /.reserved/raw prefix on any of the source and target paths.

If the /.reserved/rawprefix is specified on only a subset of the source and target paths, an error will be displayed and a non-0 exit code returned.